set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode='nonstrict';
set hive.exec.max.dynamic.partitions=1000;
set hive.exec.max.dynamic.partitions.pernode=1000;
with t_waybill as
         (select input_time
               , id
               , waybill_no
               , waybill_status_code
               , dispatch_staff_code
               , dispatch_staff_name
               , dispatch_time
               , dispatch_network_id
               , dispatch_network_code
               , dispatch_network_name
               , terminal_dispatch_code
               , is_sign
               , sign_network_id
               , sign_network_code
               , sign_network_name
               , sign_finance_code
               , sign_finance_name
               , sign_time
               , is_abnormal
               , waybill_source_code
               , waybill_source_name
               , dispatch_code
               , dispatch_name
               , is_void
               , is_refund
               , dispatch_finance_code
               , dispatch_finance_name
               , receiver_country_id
               , receiver_country_name
               , receiver_province_id
               , receiver_province_name
               , receiver_city_id
               , receiver_city_name
               , receiver_area_id
               , receiver_area_name
               , receiver_township
               , receiver_street
               , receiver_detailed_address
               , receiver_sorting_code
               , delivery_time
               , order_source_code
               , last_update_time
               , is_delete
               , distribute_finance_id
               , distribute_finance_code
               , distribute_finance_name
               , nvl(get_json_object(INVOICE_NO,'$.stationUniqueNumber'),'') stationuniquenumber -- get_json_object里面的数据大小写不能忽略 否则get不到key对应的value
               , row_number() OVER (PARTITION BY waybill_no ORDER BY last_update_time DESC) bill_day_rank
          from (select input_time
                     , id
                     , waybill_no
                     , waybill_status_code
                     , dispatch_staff_code
                     , dispatch_staff_name
                     , dispatch_time
                     , dispatch_network_id
                     , dispatch_network_code
                     , dispatch_network_name
                     , terminal_dispatch_code
                     , is_sign
                     , sign_network_id
                     , sign_network_code
                     , sign_network_name
                     , sign_finance_code
                     , sign_finance_name
                     , sign_time
                     , is_abnormal
                     , waybill_source_code
                     , waybill_source_name
                     , dispatch_code
                     , dispatch_name
                     , is_void
                     , is_refund
                     , dispatch_finance_code
                     , dispatch_finance_name
                     , receiver_country_id
                     , receiver_country_name
                     , receiver_province_id
                     , receiver_province_name
                     , receiver_city_id
                     , receiver_city_name
                     , receiver_area_id
                     , receiver_area_name
                     , receiver_township
                     , receiver_street
                     , receiver_detailed_address
                     , receiver_sorting_code
                     , delivery_time
                     , order_source_code
                     , last_update_time
                     , is_delete
                     , distribute_finance_id
                     , distribute_finance_code
                     , distribute_finance_name
                     , INVOICE_NO
                from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
                where dt >= date_add('{{ execution_date | cst_ds }}', -60)
                  and dt <= '{{ execution_date | cst_ds }}'
                  and LENGTH(waybill_no) <= 15
                  and split(regexp_replace(terminal_dispatch_code, ' ', ','), ',')[0] != ''
                  and split(regexp_replace(terminal_dispatch_code, ' ', ','), ',')[0] != '#N/A'
                  and receiver_detailed_address is not null
                  and is_delete = 1
                  and is_sign = 1
                  and is_abnormal = 0
                  and is_void = 0
                  and is_refund in (0, 2)
                  and to_date(sign_time) = '{{ execution_date | cst_ds }}'
                  and sign_network_code is not null
                ) x),
     t_waybill_distinct as (select input_time
                                 , id
                                 , waybill_no
                                 , waybill_status_code
                                 , dispatch_staff_code
                                 , dispatch_staff_name
                                 , dispatch_time
                                 , dispatch_network_id
                                 , dispatch_network_code
                                 , dispatch_network_name
                                 , terminal_dispatch_code
                                 , is_sign
                                 , sign_network_id
                                 , sign_network_code
                                 , sign_network_name
                                 , sign_finance_code
                                 , sign_finance_name
                                 , sign_time
                                 , is_abnormal
                                 , waybill_source_code
                                 , waybill_source_name
                                 , dispatch_code
                                 , dispatch_name
                                 , is_void
                                 , is_refund
                                 , dispatch_finance_code
                                 , dispatch_finance_name
                                 , receiver_country_id
                                 , receiver_country_name
                                 , case
                                       when (receiver_province_id is null) then 0
                                       else receiver_province_id
             end                                                               as receiver_province_id
                                 , receiver_province_name
                                 , case
                                       when (receiver_city_id is null) then 0
                                       else receiver_city_id
             end                                                               as receiver_city_id
                                 , receiver_city_name
                                 , case
                                       when (receiver_area_id is null) then 0
                                       else receiver_area_id
             end                                                               as receiver_area_id
                                 , receiver_area_name
                                 , case
                                       when (receiver_township is null) then ""
                                       else receiver_township
             end                                                               as receiver_township
                                 , case
                                       when (receiver_street is null) then ""
                                       else receiver_street
             end                                                               as receiver_street
                                 , regexp_replace(regexp_replace(regexp_replace(regexp_replace(regexp_replace(
                                                                                                       regexp_replace(trim(receiver_detailed_address), '"', ''),
                                                                                                       '·', ''), '（',
                                                                                               ''), '）', ''), '——',
                                                                 ''), '，', '') as receiver_detailed_address
                                 , receiver_sorting_code
                                 , delivery_time
                                 , order_source_code
                                 , last_update_time
                                 , is_delete
                                 , distribute_finance_id
                                 , distribute_finance_code
                                 , distribute_finance_name
                                 , stationUniqueNumber
                            from t_waybill
                            where bill_day_rank = 1),
     table_second as
         (SELECT *
          FROM jms_dim.dim_mysql_lmdm_sys_second_code
          WHERE is_enable = 1
            AND is_delete = 1),
     table_tail as
         (select *
          from jms_dim.dim_mysql_lmdm_sys_tail_dispatch_area
          where is_enable = 1
            and is_delete = 1),
     table_tail_staff as
         (select tail_id, tail_code, staff_code, staff_name
          from jms_dim.dim_mysql_lmdm_sys_tail_dispatch_area_staff
          group by tail_id, tail_code, staff_code, staff_name),
     table_third as
         (select a.staff_code,
                 a.staff_name,
                 b.id,
                 b.code,
                 b.name,
                 b.type,
                 b.network_id,
                 b.network_code
          from table_tail b
                   left join table_tail_staff a
                             on a.tail_id = b.id),
     t_address as (SELECT t_waybill.receiver_province_id,
                          t_waybill.receiver_province_name,
                          t_waybill.receiver_city_id,
                          t_waybill.receiver_city_name,
                          t_waybill.receiver_area_id,
                          t_waybill.receiver_area_name,
                          t_waybill.receiver_township,
                          t_waybill.receiver_street,
                          t_waybill.receiver_detailed_address,
                          0                                                                         AS send_counts,
                          1                                                                         AS receive_counts,
                          t_waybill.sign_time,
                          md5(CONCAT(t_waybill.receiver_province_id, t_waybill.receiver_city_id,
                                     t_waybill.receiver_area_id,
                                     trim(t_waybill.receiver_township), trim(t_waybill.receiver_street),
                                     trim(t_waybill.receiver_detailed_address)))                    AS md5_id,
                          t_waybill.sign_network_id                                                 as receiver_network_id,
                          t_waybill.sign_network_code                                               as receiver_network_code,
                          t_waybill.sign_network_name                                               as receiver_network_name,
                          0                                                                         as first_code_id,
                          0                                                                         as second_code_id,
                          0                                                                         as third_code_id,
                          split(regexp_replace(t_waybill.terminal_dispatch_code, ' ', ','), ',')[0] as first_code,
                          t_second.code                                                             as second_code,
                          case
                              when (t_third.code is null) then ''
                              else t_third.code
                              end                                                                   as third_code,
                          case
                              when (split(t_waybill.receiver_sorting_code, ',')[1] is null) then 0.0
                              else split(t_waybill.receiver_sorting_code, ',')[1]
                              end                                                                   as lng,
                          case
                              when (split(t_waybill.receiver_sorting_code, ',')[0] is null) then 0.0
                              else split(t_waybill.receiver_sorting_code, ',')[0]
                              end                                                                   as lat,
                          1                                                                         as lon_lat_from,
                          2                                                                         AS address_type,
                          t_waybill.receiver_sorting_code                                           as errcodeenum,
                          t_waybill.waybill_no,
                          t_waybill.stationuniquenumber
                   FROM t_waybill_distinct t_waybill
                            left JOIN
                        table_second t_second ON t_waybill.sign_network_code = t_second.network_code
                            left join
                        table_third t_third
                        on t_waybill.dispatch_staff_code = t_third.staff_code
                            and t_waybill.dispatch_staff_name = t_third.staff_name
                            and t_waybill.sign_network_code = t_third.network_code
                   WHERE t_second.code IS NOT NULL),
     t_address_distinct as (select *,
                                   row_number() over (partition by md5_id order by sign_time desc) rnk
                            from t_address),
-- insert overwrite table jms_dwd.dwd_cn_address_warehouse_day_dt partition (dt = date_add(`current_date`(),-1))
     tmp as (select receiver_province_id      province_id,
                    receiver_province_name    province_name,
                    receiver_city_id          city_id,
                    receiver_city_name        city_name,
                    receiver_area_id          area_id,
                    receiver_area_name        area_name,
                    receiver_township         township,
                    receiver_street           street,
                    receiver_detailed_address detailed_address,
                    send_counts               send_counts,
                    receive_counts            receive_counts,
                    sign_time                 update_timestamp,
                    md5_id                    md5_id,
                    receiver_network_id       network_id,
                    receiver_network_code     network_code,
                    receiver_network_name     network_name,
                    first_code_id             first_code_id,
                    second_code_id            second_code_id,
                    third_code_id             third_code_id,
                    first_code                first_code,
                    second_code               second_code,
                    third_code                third_code,
                    lng                       lng,
                    lat                       lat,
                    lon_lat_from              lon_lat_from,
                    address_type              address_type,
                    errcodeenum               errcodeenum,
                    waybill_no                waybill_no,
                    stationuniquenumber                stationuniquenumber
             from t_address_distinct
             where rnk = 1)
insert overwrite table jms_dm.dm_cn_address_warehouse_day_rule_dt_parquet partition (dt ='{{ execution_date | cst_ds }}')
select province_id,
       province_name,
       city_id,
       city_name,
       area_id,
       area_name,
       case
           when (township is null) then ""
           else township
           end                                                       as township,
       case
           when (street is null) then ""
           else street
           end                                                       as street,
       detailed_address,
       send_counts,
       receive_counts,
       update_timestamp,
       md5_id,
       network_id,
       network_code,
       network_name,
       first_code_id,
       second_code_id,
       third_code_id,
       first_code,
       second_code,
       third_code,
       lng,
       lat,
       lon_lat_from,
       address_type,
       case
           when (errcodeenum is null) then ""
           else errcodeenum
           end                                                       as errcodeenum,
       concat(province_name, city_name, area_name, detailed_address) as full_address,
       waybill_no,
       stationuniquenumber
from tmp
    distribute by pmod(hash(rand()), 399);